// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_SCANNER_HPP_
#define _XBASE_SCANNER_HPP_

#include <string>
#include <vector>

#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/intrusive_ptr.hpp>
#include <boost/system/error_code.hpp>

#include "sync_handler.hpp"
#include "location_cache.hpp"
#include "client_options.hpp"
#include "session_manager.hpp"
#include "io_service_pool.hpp"
#include "async_module_base.hpp"

namespace xbase
{
	class scanner
		: public async_module_base<io_service_pool::user_level, io_service_pool::xbase_level>
	{
	public:
		typedef async_module_base<io_service_pool::user_level, io_service_pool::xbase_level> base_type;
		typedef scanner this_type;
		typedef boost::intrusive_ptr<this_type> ptr_type;

		typedef boost::asio::ip::tcp::endpoint endpoint_type;

		typedef boost::system::error_code error_code;

		typedef boost::function<void(row_result::ptr_type, error_code)> handler_type;

		typedef basic_sync_handler<row_result::ptr_type> sync_handler_type;
	public:
		scanner(io_service_pool &pool, session_manager &sm, location_cache &lc, const scan_info	&si)
			: base_type(pool),
			m_location_cache(lc),
			m_session_manager(sm),
			m_started(false),
			m_stopped(false),
			m_include_cur_row(true),
			m_io_service_pool(pool),
			m_options(pool.options()),
			m_retry_times(0)
		{
			m_table_name = si.m_table_name;
			m_start_row = si.m_start_row;
			m_end_row = si.m_end_row;

			m_tablet_scan_info.m_column_names = si.m_column_names;
			m_tablet_scan_info.m_latest_time_stamp = si.m_latest_time_stamp;
			m_tablet_scan_info.m_num_row = si.m_num_row;
			m_tablet_scan_info.m_oldest_time_stamp = si.m_oldest_time_stamp;
			m_tablet_scan_info.m_options = si.m_options;
			m_tablet_scan_info.m_row_count = si.m_row_count;
			m_tablet_scan_info.m_version_number = si.m_version_number;
		}

		void async_get_next_row(handler_type handler)
		{
			m_handler = handler;

			if (!m_started)
			{
				m_cur_row = m_start_row;
				m_started = true;

				// 扫描未开始，查询起始行所在的range
				do_location_cache_search(m_start_row, false);
			}
			else if (m_stopped)
			{
				// 扫描已经结束
				dispatch(row_result_ptr(), xbase::error::eos);
			}
			else
			{
				do_async_get_next_row();
			}
		}

		error_code get_next_row(row_result::ptr_type &row_ptr)
		{
			sync_handler_type sh;
			async_get_next_row(sh.make_handler());
			sh.wait();

			row_ptr = sh.get_row_result_ptr();
			error_code ec = sh.get_error_code();

			return ec;
		}

		void dump_stat(scan_stat &upper_stat)
		{
			if (m_tablet_scanner_ptr)
			{
				m_tablet_scanner_ptr->dump_stat(upper_stat);
			}
		}

		bool is_stopped()
		{
			return m_stopped;
		}

	private:
		void on_get_meta(location_cache_value meta, const error_code& ec);

		void on_get_next_row(row_result::ptr_type row_ptr, const error_code &ec);

		void do_retry();

		void on_fail(const error_code &ec);

		void on_error(const error_code &ec);

		void do_new_tablet_scan(const endpoint_type &ep);

		void do_async_get_next_row();

		void do_location_cache_search(std::string &request_row, bool hard);

		void dispatch(row_result_ptr row_ptr, error_code ec);
	protected:
		ptr_type this_ptr() {
			return ptr_type(this);
		}

	private:
		location_cache &m_location_cache;
		session_manager &m_session_manager;
		boost::shared_ptr<boost::asio::deadline_timer> m_timer;

		bool m_started;
		bool m_stopped;
		bool m_include_cur_row;
		tablet_scanner_ptr m_tablet_scanner_ptr;
		tablet_scan_info m_tablet_scan_info;

		std::string m_table_name;
		std::string m_start_row;
		std::string m_end_row;
		std::string m_cur_row;

		handler_type m_handler;
		uint64_t m_cur_meta_got_time;

		io_service_pool &m_io_service_pool;
		const client_options &m_options;
		unsigned m_retry_times;

		location_cache_value m_meta_data;
	};

	typedef scanner::ptr_type scanner_ptr;
}

#endif

